package com.atguigu.flink.day07;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;

/**
 * @author Felix
 * @date 2023/12/8
 * 需求：统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次
 * 选型
 *      滑动窗口
 *          窗口大小:10s
 *          滑动步长:5s
 * 实现思路：
 *      获取全部数据
 *      开启全窗口
 *      对窗口数据做全量计算
 *          定义Map存放每一个水位值对应的次数
 *          通过list进行排序
 *          将数据输出到下游
 */
public class Flink03_topn_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner(
                        new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                return ws.getTs()*1000;
                            }
                        }
                    )
            );
        //开窗
        AllWindowedStream<WaterSensor, TimeWindow> windowDS = wsDS
            .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //对窗口数据进行处理
        windowDS.process(
            new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                @Override
                public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                    //定义一个map集合，用于存放每一个水位出现的次数
                    Map<Integer,Integer> vcCountMap = new HashMap<>();
                    //遍历窗口中的所有元素，将水位放到map中管理起来
                    for (WaterSensor ws : elements) {
                        Integer vc = ws.vc;
                        //判断map中是否包含当前水位值
                        if(vcCountMap.containsKey(vc)){
                            //说明当前水位值已经出现过了，在原来出现次数的基础上+1
                            vcCountMap.put(vc,vcCountMap.get(vc) + 1);
                        }else{
                            //说明当前水位值还没有出现过了，将其放到map中，计数1
                            vcCountMap.put(vc,1);
                        }
                    }
                    //定义一个list集合 用于排序
                    List<Tuple2<Integer,Integer>> vcCountList = new ArrayList<>();
                    Set<Map.Entry<Integer, Integer>> entrySet = vcCountMap.entrySet();
                    for (Map.Entry<Integer, Integer> entry : entrySet) {
                        vcCountList.add(Tuple2.of(entry.getKey(),entry.getValue()));
                    }

                    //排序
                    vcCountList.sort((o1, o2) -> o2.f1 - o1.f1);

                    //取出现次数排名前2
                    StringBuilder outStr = new StringBuilder();
                    outStr.append("===============================\n");
                    for (int i = 0; i < Math.min(2,vcCountList.size()); i++) {
                        Tuple2<Integer, Integer> vcCountTuple = vcCountList.get(i);
                        Integer vc = vcCountTuple.f0;
                        Integer count = vcCountTuple.f1;
                        outStr.append("Top"+(i + 1)+"\n");
                        outStr.append("VC:"+vc+",COUNT:"+count+"\n");
                        String stt = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss");
                        String edt = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss");
                        outStr.append("当前窗口时间["+stt+"~"+edt + ")\n");
                        outStr.append("===============================\n");
                    }

                    out.collect(outStr.toString());
                }
            }
        ).print();

        env.execute();
    }
}
